package report

import Configer.Config
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

//广告项目用户打标签
object TagsDMP {
  def main(args: Array[String]): Unit = {
    //sparkContext
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getName}")
    conf.setMaster("local[*]")
    conf.set("spark.serializer", Config.serializer)

    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)
    //读取数据
    val dictLine = sc.textFile("C:\\Users\\44323\\Desktop\\资料PDF\\app_dict.txt")
    val stopLine = sc.textFile("C:\\Users\\44323\\Desktop\\资料PDF\\stopwords.txt")
    val dataFrame = sQLContext.read.parquet(Config.parquetPath)

    //广播变量
    val dictMap = dictLine.map(_.split("\t", -1)).filter(_.length >= 5).map(arr => (arr(4), arr(1))).collectAsMap()
    val dictBro = sc.broadcast(dictMap)

    val stopMap = stopLine.map((_, 1)).collectAsMap()
    val stopBro = sc.broadcast(stopMap)
    //处理数据
    dataFrame.filter(
      """
        |imei != "" or mac != "" or idfa != "" or openudid != "" or androidid != "" or
        |imeimd5 != "" or macmd5 != "" or idfamd5 != "" or openudidmd5 != "" or androididmd5 != "" or
        |imeisha1 != "" or macsha1 != "" or idfasha1 != "" or openudidsha1 != "" or androididsha1 != ""
      """.stripMargin)
      .map(row => {
      var list = List[(String, Int)]()
      //广告位
      val adspacetype = row.getAs[Int]("adspacetype")
      if (adspacetype >= 10) {
        list :+= ("LC"+adspacetype, 1)
      } else if (adspacetype < 10 && adspacetype > 0) {
        list :+= ("LC" + adspacetype, 1)
      }

      val adspacetypename = row.getAs[String]("adspacetypename")
      if (StringUtils.isNotEmpty(adspacetypename)) {
        list :+= ("LN" + adspacetypename, 1)
      }

      //app
      var appName = row.getAs[String]("appname")
      val appId = row.getAs[String]("appid")
      if (StringUtils.isEmpty(appName)) {
        if (StringUtils.isNotEmpty(appId)) {
          appName = dictBro.value.getOrElse(appId, appId)
          list :+= ("APP" + appName, 1)
        }
      } else {
        list :+= ("APP" + appName, 1)
      }

      //渠道
      val adplatformproviderid = row.getAs[Int]("adplatformproviderid")
      if (adplatformproviderid > 0) {
        list :+= ("CN" + adplatformproviderid, 1)
      }

      //设备 操作系统  联网方式  运营商
      val client = row.getAs[Int]("client")
      val networkmannername = row.getAs[String]("networkmannername")
      val ispname = row.getAs[String]("ispname")

      if (client == 1) {
        list :+= ("D00010001", 1)
      } else if (client == 2) {
        list :+= ("D00010002", 1)
      } else if (client == 3) {
        list :+= ("D00010003", 1)
      } else {
        list :+= ("D00010004", 1)
      }

      if ("WIFI".equals(networkmannername.toUpperCase)) {
        list :+= ("D00020001", 1)
      } else if ("4G".equals(networkmannername.toUpperCase)) {
        list :+= ("D00020002", 1)
      } else if ("3G".equals(networkmannername.toUpperCase)) {
        list :+= ("D00020003", 1)
      } else if ("2G".equals(networkmannername.toUpperCase)) {
        list :+= ("D00020004", 1)
      } else {
        list :+= ("D00020005", 1)
      }
      ispname match {
        case "移动" => list :+= ("D00030001", 1)
        case "联通" => list :+= ("D00030002", 1)
        case "电信" => list :+= ("D00030003", 1)
        case _ => list :+= ("D00030004", 1)
      }

      //关键字
      val keywords = row.getAs[String]("keywords")
      keywords.split("\\|", -1).filter(tp => tp.length >= 3 && tp.length <= 8 && !stopBro.value.contains(tp))
        .foreach(k => list :+= ("K" + k, 1))

      //地域标签
      val pname = row.getAs[String]("provincename")
      val cname = row.getAs[String]("cityname")
      if (StringUtils.isNotEmpty(pname)) {
        list :+= ("ZP" + pname, 1)
      }
      if (StringUtils.isNotEmpty(cname)) {
        list :+= ("ZC" + cname, 1)
      }

      //userid
      var userid = ""
      row match {
        case v if StringUtils.isNotEmpty(row.getAs[String]("imei")) => userid = "imei:" + row.getAs[String]("imei")
        case v if StringUtils.isNotEmpty(row.getAs[String]("mac")) => userid = "mac:" + row.getAs[String]("mac")
        case v if StringUtils.isNotEmpty(row.getAs[String]("idfa")) => userid = "idfa:" + row.getAs[String]("idfa")
        case v if StringUtils.isNotEmpty(row.getAs[String]("openudid")) => userid = "openudid:" + row.getAs[String]("openudid")
        case v if StringUtils.isNotEmpty(row.getAs[String]("androidid")) => userid = "androidid:" + row.getAs[String]("androidid")

        case v if StringUtils.isNotEmpty(row.getAs[String]("imeimd5")) => userid = "imeimd5:" + row.getAs[String]("imeimd5")
        case v if StringUtils.isNotEmpty(row.getAs[String]("macmd5")) => userid = "macmd5:" + row.getAs[String]("macmd5")
        case v if StringUtils.isNotEmpty(row.getAs[String]("idfamd5")) => userid = "idfamd5:" + row.getAs[String]("idfamd5")
        case v if StringUtils.isNotEmpty(row.getAs[String]("openudidmd5")) => userid = "openudidmd5:" + row.getAs[String]("openudidmd5")
        case v if StringUtils.isNotEmpty(row.getAs[String]("androididmd5")) => userid = "androididmd5:" + row.getAs[String]("androididmd5")

        case v if StringUtils.isNotEmpty(row.getAs[String]("imeisha1")) => userid = "imeisha1:" + row.getAs[String]("imeisha1")
        case v if StringUtils.isNotEmpty(row.getAs[String]("macsha1")) => userid = "macsha1:" + row.getAs[String]("macsha1")
        case v if StringUtils.isNotEmpty(row.getAs[String]("idfasha1")) => userid = "idfasha1:" + row.getAs[String]("idfasha1")
        case v if StringUtils.isNotEmpty(row.getAs[String]("openudidsha1")) => userid = "openudidsha1:" + row.getAs[String]("openudidsha1")
        case v if StringUtils.isNotEmpty(row.getAs[String]("androididsha1")) => userid = "androididsha1:" + row.getAs[String]("androididsha1")
      }
      (userid,list)
    }).reduceByKey{
      (l1,l2)=>(l1++l2).groupBy(_._1)//.mapValues(v=>v.length).toList
            //.mapValues(v=>v.map(_._2).sum).toList
          .mapValues(v=>v.foldLeft(0)(_+_._2)).toList
    }.coalesce(1).saveAsTextFile("C:\\Users\\44323\\Desktop\\资料PDF\\sss2")

    sc.stop()
  }
}
